package com.cmdfut.rxjava.rxjava;

//指定被观察者在哪个线程运行的被观察者
public class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
    public ObservableSubscribeOn(ObservableSource<T> source) {
        super(source);
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        final SubscribeOnObserver parent = new SubscribeOnObserver(observer);
        //告诉下游观察者订阅成功
        source.subscribe(parent);
    }

    static final class SubscribeOnObserver<T> implements Observer<T>, Disposable {

        final Observer<? super T> actual;
        Disposable<T> disposable;

        public SubscribeOnObserver(Observer<? super T> actual) {
            this.actual = actual;
        }

        @Override
        public void dispose(boolean bool) {
            this.disposable.dispose(bool);
        }

        @Override
        public boolean isDisposed() {
            return this.disposable.isDisposed();
        }

        @Override
        public void onSubscribe(Disposable d) {
            this.disposable = d;
        }

        @Override
        public void onNext(T t) {
            this.actual.onNext(t);
        }

        @Override
        public void onError(Throwable e) {
            this.actual.onError(e);
        }

        @Override
        public void onComplete() {
            this.actual.onComplete();
        }
    }
}
